package studio.raptor.extractor.segment.server;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.ApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Scope;
import org.springframework.stereotype.Component;
import studio.raptor.extractor.common.holder.ExecutorHolder;
import studio.raptor.extractor.segment.config.ExtractorConfig;
import studio.raptor.extractor.segment.config.ServerConfig;
import studio.raptor.extractor.segment.entity.BuildJsonResult;
import studio.raptor.extractor.segment.entity.EventDataInfo;
import studio.raptor.extractor.segment.entity.EventMeta;
import studio.raptor.extractor.segment.service.EventMetaService;
import studio.raptor.extractor.segment.service.EventNodeService;
import studio.raptor.extractor.segment.task.*;

import javax.annotation.PostConstruct;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.LinkedBlockingQueue;

/**
 * Created by liuyong8 on 2017/7/12.
 */
@Component
@Scope("prototype")
public class ExtractorServer {
  private static final Logger log = LoggerFactory.getLogger(ExtractorServer.class);

  @Autowired
  ApplicationContext context;

  // 存放带抽文的eventmeta队列
  private BlockingQueue<EventMeta> eventQueue;

  // 存放节点报文
  private BlockingQueue<EventDataInfo> evenDataQueue;

  // 组装队列
  private BlockingQueue<BuildJsonResult> buildJsonQueue;

  private ServerConfig serverConfig;

  public static final int coreSize = Runtime.getRuntime().availableProcessors();

  public static final int maxSize = Runtime.getRuntime().availableProcessors() * 40;

  @Autowired
  private EventMetaService eventMetaService;
  @Autowired
  private EventNodeService eventNodeService;

  private ExecutorService extractingproducerExecutor;
  // 事件报文抽取线程池
  private ExecutorService extractingExecutor ;
  //事件激活线程池
  private ExecutorService activeExecutor;
  //节点报文线程池
  private ExecutorService assemblingExecutor;
  //event报文组装线程池
  private ExecutorService buildingExecutor;
  //节点报文组装线程池
  private ExecutorService nodeBuildingExecutor;
  //发送线程池
  private ExecutorService deliveringExecutor;


  @PostConstruct
  public void init(){
    //TODO 去读配置，设置队列大小,默认值为100
    this.eventQueue = new LinkedBlockingQueue<EventMeta>(100);
    evenDataQueue = new LinkedBlockingQueue<EventDataInfo>(100);
    buildJsonQueue = new LinkedBlockingQueue<>(100);

  }

  /**
   * 启动入口
   */
  public void startup() throws Exception{


    //TODO 设置超时时间,单位毫秒
    this.getServerConfig().setOverTime(5000L);
    this.getServerConfig().setServerUrlMap(eventNodeService.findNodeInfosByKindId(serverConfig.getKindId()));

    extractingproducerExecutor = ExecutorHolder.createSingleThreadPool("event_producter_extracting" + "_" + serverConfig.getKindId());
    extractingExecutor = ExecutorHolder.createSingleThreadPool("event_mq_extracting" + "_" + serverConfig.getKindId());
    activeExecutor = ExecutorHolder.createSingleThreadPool("event_activing" + "_" + serverConfig.getKindId());
    assemblingExecutor = ExecutorHolder.createCachedThreadPool("node_msg_assembling" + "_" + serverConfig.getKindId(), coreSize, maxSize);
    buildingExecutor = ExecutorHolder.createSingleThreadPool("event_building" + "_" + serverConfig.getKindId());
    nodeBuildingExecutor = ExecutorHolder.createCachedThreadPool("nod_building" + "_" + serverConfig.getKindId(), coreSize, maxSize);
    deliveringExecutor = ExecutorHolder.createSingleThreadPool("msg_deliver" + "_" + serverConfig.getKindId());

//    extractingExecutor.submit(new EventRecoverTask());
//    extractingproducerExecutor.submit(new EventMQProduterTask());
    extractingExecutor.submit(createMQExtractingTask());
//    activeExecutor.submit(createEventActiveTask());
    assemblingExecutor.submit(createNodeMsgAssemblingTask());
    buildingExecutor.submit(createBuilJsonTask());
    deliveringExecutor.submit(createEventDeliveringTask());

  }

  private EventDeliveringTask createEventDeliveringTask(){
    EventDeliveringTask eventDeliveringTask = context.getBean(EventDeliveringTask.class);
    eventDeliveringTask.setBuildJsonQueue(buildJsonQueue);
    eventDeliveringTask.setServerConfig(serverConfig);
    return eventDeliveringTask;
  }

  private BuildJsonTask createBuilJsonTask(){
    BuildJsonTask buildJsonTask = context.getBean(BuildJsonTask.class);
    buildJsonTask.setBuildJsonQueue(buildJsonQueue);
    buildJsonTask.setEventDataQueue(evenDataQueue);
    buildJsonTask.setNodeBuildingExecutor(nodeBuildingExecutor);
    buildJsonTask.setServerConfig(serverConfig);
    return buildJsonTask;
  }

  private NodeMsgAssemblingTask createNodeMsgAssemblingTask(){
    NodeMsgAssemblingTask nodeMsgAssemblingTask = context.getBean(NodeMsgAssemblingTask.class);
    nodeMsgAssemblingTask.setEventMetaQueue(eventQueue);
    nodeMsgAssemblingTask.setEventDataQueue(evenDataQueue);
    nodeMsgAssemblingTask.setExec(assemblingExecutor);
    nodeMsgAssemblingTask.setServerConfig(serverConfig);
    return  nodeMsgAssemblingTask;
  }

  private EventMetaService createEventMetaService(){
    EventMetaService eventMetaService = context.getBean(EventMetaService.class);
    eventMetaService.setServerConfig(serverConfig);
    eventMetaService.setEventQueue(eventQueue);
    return eventMetaService;
  }

  private EventMQExtractingTask createMQExtractingTask(){
    EventMQExtractingTask eventMQExtractingTask = context.getBean(EventMQExtractingTask.class);
    eventMQExtractingTask.setServerConfig(this.getServerConfig());
    eventMQExtractingTask.setEventQueue(eventQueue);
    return eventMQExtractingTask;
  }

  private EventActiveTask createEventActiveTask(){
    EventActiveTask eventActiveTask = context.getBean(EventActiveTask.class);
    eventActiveTask.setServerConfig(this.getServerConfig());
    eventActiveTask.setEventQueue(this.getEventQueue());
    return eventActiveTask;
  }

  /**
   * recover 入口
   */
  public void recover() throws Exception{
    eventMetaService.recover();
    // step1:query event meta status P、F number ，metaId asc


    // step1:recover event meta status P、F -> W

    // step2:put into eventQueue by order

  }

  public ServerConfig getServerConfig() {
    return serverConfig;
  }

  public void setServerConfig(ServerConfig serverConfig) {
    this.serverConfig = serverConfig;
  }

  public BlockingQueue<EventMeta> getEventQueue() {
    return eventQueue;
  }

  public BlockingQueue<EventDataInfo> getEvenDataQueue() {
    return evenDataQueue;
  }

  public BlockingQueue<BuildJsonResult> getBuildJsonQueue() {
    return buildJsonQueue;
  }
}
